#include "delivery_invoker.h"    
#include "log.h"
#include "prepare.h"
#include "singleton.h"
#include "delivery_cleaner.h"

namespace edm{
namespace prepare{
        
ssize_t write_n(int sockfd, const char* vptr, size_t n)
{
    size_t nleft;
    ssize_t nwritten;
    const char *ptr;
    
    ptr = vptr;
    nleft = n;
    while(nleft > 0)
    {
        if((nwritten = write(sockfd, ptr, nleft)) <= 0)
        {
            if(nwritten < 0 && errno == EINTR)
                nwritten = 0;
            else
                return -1;
        }
        nleft -= nwritten;
        ptr += nwritten;
    }
    
    return n;
}
ssize_t read_n(int fd, void* vptr, size_t n)
{
    size_t nleft;
    ssize_t nread;
    char *ptr;
    
    ptr = (char*)vptr;
    nleft = n;
    while(nleft > 0)
    {
        if((nread = read(fd, ptr, nleft)) < 0)
        {
            if(errno == EINTR)
                nread = 0;
            else
                return -1;
        }
        else if(nread == 0)
            break; //EOF
            
        nleft -= nread;
        ptr += nread;
    }
    
    return (n - nleft);    

}

delivery_invoker_t::delivery_invoker_t(const char* bind, const char* udp_host, int concurrency):
        m_bind_host(bind),m_udp_host(udp_host),
        m_concurrency(concurrency),
        m_started(false),m_callback_fork(0),m_callback_error(0),
        m_io_work(new boost::asio::io_service::work(m_io_service))
{
    logtrace((PREPARE, "delivery_invoker_t::delivery_invoker_t <%s:%d>", m_bind_host.c_str(), concurrency));
}

delivery_invoker_t::~delivery_invoker_t()
{
    logtrace((PREPARE, "delivery_invoker_t::~delivery_invoker_t begin ..."));
    if(m_started) stop();
    logtrace((PREPARE, "delivery_invoker_t::~delivery_invoker_t end ok."));
}
    
//!  start: starting delivery invoker service
//!  @in   void
//!  @result: int    
int delivery_invoker_t::start(const std::string& maildrop_path,
                      const std::string& maildrop_conf, 
                      callback_fork_t callback_fork,
                      callback_error_t callback_error)
{
  logtrace((PREPARE, "delivery_invoker_t::start begin..."));

  if (m_started)
  {
    logwarn((PREPARE, "delivery_invoker_t::start already started."));
    return 0;
  }
  m_maildrop_path = maildrop_path;
  m_maildrop_conf = maildrop_conf;
  m_callback_fork = callback_fork;
  m_callback_error = callback_error;
  for(int i = 0; i < m_concurrency; i++)
  {
      m_thread_group.create_thread(boost::bind(&boost::asio::io_service::run, 
                                               &m_io_service));
  }
  m_started = true;

  logtrace((PREPARE, "delivery_invoker_t::start end ok."));
  return 0;
}

//!  stop: stopping delivery invoker service, will be waitting all 
//!        delivery item to be invokedd
//!  @in   delivery_item_t: item
//!  @result: int    
int delivery_invoker_t::stop()
{
  logtrace((PREPARE, "delivery_invoker_t::stop begin..."));

  if (!m_started)
  {
    logwarn((PREPARE, "delivery_invoker_t::stop already started."));
    return 0;
  }
  
  m_io_work.reset();
  m_thread_group.join_all();
  m_started = false;
  
  logtrace((PREPARE, "delivery_invoker_t::stop end ok."));
  return 0;
}

//!  invoke: invoke a process to delivery the item
//!  @in   delivery_item_t: item
//!  @result: void
void delivery_invoker_t::maildrop(delivery_t* delivery)
{
  logtrace((PREPARE, "delivery_invoker_t::maildrop"));
  m_io_service.post(boost::bind(&delivery_invoker_t::maildrop_i, this, delivery));
}

void delivery_invoker_t::maildrop_i(delivery_t* delivery)
{
  logtrace((PREPARE, "delivery_invoker_t::maildrop_i begin..."));
  
  //! 创建管道
  int fd[2];
  if(pipe(fd) < 0)
  {
    delivery->state = delivery_t::COOKED_TEMPERR;
    if(m_callback_error) m_callback_error(delivery);
    logerror((PREPARE, "delivery_invoker_t::maildrop_i => pipe create failed:%s.", strerror(errno)));
    return;      
  }
  
  //! fork a child process to delivery
  pid_t pid = fork();
  if(pid < 0)
  {
    delivery->state = delivery_t::COOKED_TEMPERR;
    if(m_callback_error) m_callback_error(delivery);
    logerror((PREPARE, "delivery_invoker_t::maildrop_i => fork failed:%s", strerror(errno)));
    return;
  }
  
  //! child process
  if(pid == 0)
  {
      //! copy pipe fd to stdin
      close(fd[1]);
      if(dup2(fd[0], 0))
      {
          logerror((PREPARE, "delivery_invoker_t::maildrop_i => dup2 failed:%s", strerror(errno)));
          exit(-1);          
      }
      //! child process should exit internal side
      if(m_maildrop_path.empty())
      {
          logerror((PREPARE, "delivery_invoker_t::maildrop_i => maildrop path empty"));
          exit(-1);
      }
      
      char* (args[14]);
      args[0] = (char*)m_maildrop_path.c_str();
      args[1] = "-b";
      args[2] = (char*)m_bind_host.c_str();
      args[3] = "-u";
      args[4] = (char*)m_udp_host.c_str();      
      args[5] = "-f";
      args[6] = (char*)delivery->from_addr.c_str();
      args[7] = "-t";
      args[8] = (char*)delivery->recipient.c_str();
      args[9] = "-T";
      args[10] = "300";

      if(!m_maildrop_conf.empty())
      {
        args[11] = "-O";
        args[12] = (char*)m_maildrop_conf.c_str();
        args[13] = 0;
      }
      else
      {
        args[11] = 0;
        args[12] = 0;
        args[13] = 0;
      }    
      
      execv(*args,args);
      //! exit external side
      exit(-1);
  }
  
  //! pid > 0, parent process, told cleaner to clean the child process
  if(m_callback_fork) m_callback_fork(pid, delivery);
  
  close(fd[0]); //! close read end
  if(write_n(fd[1], delivery->cooked_message.c_str(), delivery->cooked_message.length()) < 0)
  {
      logerror((PREPARE, "delivery_invoker_t::maildrop_i parent write to pipe failed:%s.", strerror(errno)));
      close(fd[1]);
      return;
  }

  close(fd[1]);
  logtrace((PREPARE, "delivery_invoker_t::maildrop_i end ok."));
}

}}
